package com.snts.synchronization.core;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import sun.security.action.GetLongAction;

import com.snts.synchronization.core.Change.Type;
import com.snts.synchronization.core.SyncTable.JoinRoute;
import com.snts.synchronization.garbage.Utils;
import com.snts.synchronization.schema.SchemaUpdatesGenerator;
import com.snts.synchronization.utils.OutputFields;
import com.snts.synchronization.utils.OutputFields.Printer;
import com.snts.synchronization.utils.QueryHelper;

public class TableLog {

	private static final Log log = LogFactory.getLog(TableLog.class);

	private SyncTable table;
	private long lastLogTime = 0L;

	public static final String ID_COL_NAME = "id";
	
	public static final String UPDATE_TYPE_COL_NAME = "update_type";
	public static final String UPDATE_TIME_COL_NAME = "update_time";
	public static final String ROW_ID_COL_NAME = "row_id";
	public static final String COLS_COL_NAME = "cols";
	public static final String VALUES_COL_NAME = "values";
	public static final String TRANSACTION_ID_COL_NAME = "transaction_id";

	private static final String[] columns = new String[] { ID_COL_NAME, UPDATE_TYPE_COL_NAME, UPDATE_TIME_COL_NAME, ROW_ID_COL_NAME,
			COLS_COL_NAME, VALUES_COL_NAME, TRANSACTION_ID_COL_NAME };

	private String sql;

	public TableLog(SyncTable table) {
		super();
		this.table = table;
	}
	
	private String getSql(){
		if(sql == null){
			boolean quoteIdentifier = table.getSynchronization().isQuoteIdentifier();
		}
		return sql;
	}

	private String getSql(boolean byRestaurant) {
		if (sql == null) {
			boolean quoteIdentifier = table.getSynchronization().isQuoteIdentifier();
			if (!byRestaurant) {
				StringBuffer buf = new StringBuffer("select *");
				buf.append(" from ");
				buf.append(table.getLogTable().getQualifiedName());
				buf.append(" where ");
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(UPDATE_TIME_COL_NAME);
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(" > ?");
				sql = buf.toString();
			} else {
				QueryHelper helper = new QueryHelper();
				JoinRoute joinRoute = table.getJoinRoute();

				StringBuffer buf = new StringBuffer("select ");
				buf.append(helper.getTableAlias(table.getLogTable()));
				buf.append(".* ");
				buf.append(" from ");
				buf.append(table.getLogTable().getQualifiedName());
				buf.append(" ");
				buf.append(helper.getTableAlias(table.getLogTable()));
				buf.append(",");
				buf.append(table.getQualifiedName());
				buf.append(" ");
				buf.append(helper.getTableAlias(table));

				List<TableReference> subList = joinRoute.getRoute().subList(0, joinRoute.getRoute().size() - 1);
				for (TableReference tr : subList) {
					buf.append(", ");
					buf.append(tr.getQualifiedName());
					buf.append(" ");
					buf.append(helper.getTableAlias(tr));
				}

				buf.append(" where ");
				buf.append(helper.getTableAlias(table));
				buf.append('.');
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(table.getSynchronization().getPkColumnName());
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append('=');
				buf.append(helper.getTableAlias(table.getLogTable()));
				buf.append('.');
				buf.append("row_id");
				buf.append(" AND ");
				BaseTable lastTable = table;
				for (int i = 0; i < subList.size(); i++) {
					TableReference tr = subList.get(i);
					buf.append(helper.getTableAlias(tr));
					buf.append('.');
					buf.append(table.getSynchronization().getPkColumnName());
					buf.append('=');
					buf.append(helper.getTableAlias(lastTable));
					buf.append('.');
					if(quoteIdentifier){
						buf.append('"');
					}
					buf.append(tr.getForeignKey());
					if(quoteIdentifier){
						buf.append('"');
					}
					buf.append(" and ");
					lastTable = tr;
				}
				buf.append(helper.getTableAlias(table.getLogTable()));
				buf.append('.');
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(UPDATE_TIME_COL_NAME);
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(" > ?");
				buf.append(" and ");

				buf.append(helper.getTableAlias(lastTable));
				buf.append('.');
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(joinRoute.getLastJoinTable().getRestIdColumn());
				if(quoteIdentifier){
					buf.append('"');
				}
				buf.append(" = UUID(?)");
				sql = buf.toString();
				
			}
		}
		return sql;
	}
	
	private List<Change> createChangesFromResultSet(ResultSet rs) throws SQLException {
		ArrayList<Change> changes = new ArrayList<Change>();
		int count = 0;
		while (rs.next()) {
			String type = rs.getString("update_type").toUpperCase();
			Change ch = new Change("I".equals(type) ? Type.INSERT : "U".equals(type) ? Type.UPDATE : Type.DELETE);
			ch.setTable(table);
			ch.setId(new UUID(rs.getString(ID_COL_NAME)));
			ch.setRowId(new UUID(rs.getString(ROW_ID_COL_NAME)));
			ch.setTime(new TimeStampWithTimeZone(rs.getString(UPDATE_TIME_COL_NAME)));
			String transactionId = rs.getString(TRANSACTION_ID_COL_NAME);
			if (transactionId != null) {
				ch.setTransactionId(new UUID(transactionId));
			}
			String[] values = Utils.parseLogValues(rs.getString(VALUES_COL_NAME)).toArray(new String[]{});
			int idx = 0;
			for (String str : Utils.parseLogValues(rs.getString(COLS_COL_NAME))) {
				Column col = table.getColumnByName(str);
				if (col != null) {
					ChangeColumn chc = ch.addColumn(col);
					chc.setAsString(values[idx++]);
				}
			}
			changes.add(ch);
			lastLogTime = rs.getTimestamp(UPDATE_TIME_COL_NAME).getTime();
			count ++;
		}
		System.out.println(count + " changes received from database");
		return changes;
	}
	
	

	public List<Change> getChanges(Connection conn) throws SQLException {
		PreparedStatement pstmt = conn.prepareStatement(getSql(false));
		pstmt.setTimestamp(1, new Timestamp(lastLogTime));
		ResultSet rs = pstmt.executeQuery();
		return createChangesFromResultSet(rs);
	}

	public List<Change> getChanges(Connection conn, UUID restaurantId) throws SQLException {
		String sql = getSql(true);
		log.debug("getChanges(sql=" + sql + ")");
		PreparedStatement pstmt = conn.prepareStatement(sql);
		pstmt.setTimestamp(1, new Timestamp(lastLogTime));
		pstmt.setString(2, restaurantId.toString());
		ResultSet rs = pstmt.executeQuery();
		return createChangesFromResultSet(rs);
	}

	public void deleteChange(Connection conn, Change change) throws SQLException {
		String sql = "DELETE FROM synchronization.\"" + table.getLogTableName() + "\" WHERE id=?";
		PreparedStatement pstmt = conn.prepareStatement(sql);
		pstmt.setObject(1, java.util.UUID.fromString(change.getRowId().toString()));
		pstmt.executeUpdate();
	}

	public static void main(String[] args) throws ClassNotFoundException, SQLException, InterruptedException {
		
		Class.forName("org.postgresql.Driver");
		Connection conn = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:5432/rgenius", "postgres", "123");
		Synchronization synchronization = new Synchronization();
		synchronization.load(conn.getMetaData());
		
		SchemaUpdatesGenerator schemaUpdatesGenerator = new SchemaUpdatesGenerator();
		schemaUpdatesGenerator.installUpdates(conn, synchronization);
		
		SyncTable syncTable = synchronization.getTables().get(0);
		
		OutputFields of = new OutputFields();
		of.addField(20);
		of.addField(20);
		of.addField(20);
		of.addField(40);
		of.addField(70);
	
		while (true) {
			System.out.println("Polling database for changes...");
			List<Change> changes = syncTable.getTableLog().getChanges(conn, new UUID("9c9cac82-ca67-11e1-aa21-1748b17a899a"));
			Printer printer = of.getPrinter(System.out);
			for (Change ch : changes) {
				printer.print(ch.getTable().getName());
				printer.print(ch.getType().name());
				String str = "";
				for (ChangeColumn c : ch.getColumns()){
					str += c.getColumn().getName();
					str +=",";
				}
				printer.print(str);
				str = "";
				for (Object val : ch.getValues()) {
					str += val.toString();
					str += ",";
				}
				printer.print(str);
				printer.print(ch.getTransactionId().toString());
			}
			printer.newLine();
			Thread.sleep(5 * 1000);
		}		
		

/*		
 		Class.forName("org.postgresql.Driver");
		Connection conn = DriverManager.getConnection("jdbc:postgresql://127.0.0.1:5432/Restaurant", "postgres", "123");
		Synchronization synchronization = new Synchronization();
		synchronization.load(conn.getMetaData());
		SyncTable syncTable = synchronization.getTables().get(0);
		while (true) {
			System.out.println("Polling database for changes...");
			List<Change> changes = syncTable.getTableLog().getChanges(conn);
			for (Change ch : changes) {
				System.out.print(ch.getTable().getName());
				System.out.print(" ");
				System.out.print(ch.getType().name());
				System.out.print(" ");
				for (Column c : ch.getColumns())
					System.out.print(c.getName() + ",");
				for (String str : ch.getValues()) {
					System.out.print(str + ",");
				}
				System.out.print(ch.getTransactionId());
				System.out.println();
			}
			Thread.sleep(5 * 1000);
		}
*/
		
	}

}
